package com.myCompany.edu.realtime.app.dws;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.myCompany.edu.realtime.app.BaseAppV1;
import com.myCompany.edu.realtime.bean.TradeSourceProvinceUserOrderBean;
import com.myCompany.edu.realtime.common.Constant;
import com.myCompany.edu.realtime.util.AtguiguUtil;
import com.myCompany.edu.realtime.util.DimUtil;
import com.myCompany.edu.realtime.util.DruidDSUtil;
import com.myCompany.edu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.sql.SQLException;
import java.time.Duration;

/*
下单次数、下单金额
 */
public class Dws14_DwsTradeSourceProvinceUserOrderWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws14_DwsTradeSourceProvinceUserOrderWindow().init(
                3014,
                2,
                "Dws14_DwsTradeSourceProvinceUserOrderWindow",
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //1、从kafka读取订单明细主题读取数据，按照订单详情id去掉重复数据
        SingleOutputStreamOperator<TradeSourceProvinceUserOrderBean> beanStream = parseToBean(stream);
        //2、按照source_id,province_id,user_id,分组开窗聚合
        SingleOutputStreamOperator<TradeSourceProvinceUserOrderBean> resultStreamWithoutDim = WindowAndAgg(beanStream);
        //3、补充维度信息
        SingleOutputStreamOperator<TradeSourceProvinceUserOrderBean> resultStreamWithDim = addDim(resultStreamWithoutDim);
        //4、写出到clickhouse
        writeToClickHouse(resultStreamWithDim);

    }

    private void writeToClickHouse(SingleOutputStreamOperator<TradeSourceProvinceUserOrderBean> resultStreamWithDim) {
        resultStreamWithDim.addSink(FlinkSinkUtil.getClickHouseSink("dws_trade_source_province_user_order_window",TradeSourceProvinceUserOrderBean.class));
    }

    private SingleOutputStreamOperator<TradeSourceProvinceUserOrderBean> addDim(SingleOutputStreamOperator<TradeSourceProvinceUserOrderBean> resultStreamWithoutDim) {
        return resultStreamWithoutDim
                .map(new RichMapFunction<TradeSourceProvinceUserOrderBean, TradeSourceProvinceUserOrderBean>() {

                    private DruidPooledConnection conn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        DruidDataSource dataSource = DruidDSUtil.getDataSource();
                        conn = dataSource.getConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        if(conn!=null){
                            conn.close();
                        }
                    }

                    @Override
                    public TradeSourceProvinceUserOrderBean map(TradeSourceProvinceUserOrderBean bean) throws Exception {
                        //1、补充base_source信息
                        //select * from table where id = ?
                        //将查询后的数据封装在JSONObject中
                        JSONObject baseSource = DimUtil.readDimFromPhoenix(conn,"dim_base_source",bean.getSourceId());
                        bean.setSourceName(baseSource.getString("SOURCE_SITE"));

                        //2、补充base_province
                        JSONObject baseProvince = DimUtil.readDimFromPhoenix(conn, "dim_base_province", bean.getProvinceId());
                        bean.setProvinceName(baseProvince.getString("NAME"));

                        return bean;
                    }
                });
    }

    private SingleOutputStreamOperator<TradeSourceProvinceUserOrderBean> WindowAndAgg(SingleOutputStreamOperator<TradeSourceProvinceUserOrderBean> beanStream) {
        return beanStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TradeSourceProvinceUserOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((bean,ts) -> bean.getTs()*1000)
                )
                .keyBy(bean -> bean.getSourceId() + "_" + bean.getProvinceId() + "_" + bean.getUserId())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TradeSourceProvinceUserOrderBean>() {
                            @Override
                            public TradeSourceProvinceUserOrderBean reduce(TradeSourceProvinceUserOrderBean bean1,
                                                                           TradeSourceProvinceUserOrderBean bean2) throws Exception {
                                bean1.setOrderCount(bean1.getOrderCount() + bean2.getOrderCount());
                                bean1.setOrderAmount(bean1.getOrderAmount().add(bean2.getOrderAmount()));
                                return bean1;
                            }
                        },
                        new ProcessWindowFunction<TradeSourceProvinceUserOrderBean, TradeSourceProvinceUserOrderBean, String, TimeWindow>() {
                            @Override
                            public void process(String key,
                                                Context ctx,
                                                Iterable<TradeSourceProvinceUserOrderBean> iterable,
                                                Collector<TradeSourceProvinceUserOrderBean> collector) throws Exception {
                                TradeSourceProvinceUserOrderBean bean = iterable.iterator().next();
                                String stt = AtguiguUtil.toDateTime(ctx.window().getStart());
                                String edt = AtguiguUtil.toDateTime(ctx.window().getEnd());
                                bean.setStt(stt);
                                bean.setEdt(edt);
                                bean.setTs(System.currentTimeMillis());
                                collector.collect(bean);
                            }
                        });

    }

    private SingleOutputStreamOperator<TradeSourceProvinceUserOrderBean> parseToBean(DataStreamSource<String> stream) {
        return stream.map(JSON::parseObject)
                .keyBy(obj -> obj.getString("id"))
                .map(new RichMapFunction<JSONObject, TradeSourceProvinceUserOrderBean>() {

                    private ValueState<String> orderInfoState;
                    private ValueState<String> orderDetailState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        orderDetailState = getRuntimeContext().getState(new ValueStateDescriptor<String>("orderDetailState", String.class));
                        orderInfoState = getRuntimeContext().getState(new ValueStateDescriptor<String>("OrderInfoState", String.class));
                    }

                    @Override
                    public TradeSourceProvinceUserOrderBean map(JSONObject object) throws Exception {
                        String provinceId = object.getString("province_id");
                        String sourceId = object.getString("source_id");
                        String userId = object.getString("user_id");
                        String id = object.getString("id");
                        String orderId = object.getString("order_id");
                        String splitFinalAmount = object.getString("split_final_amount");
                        Long ts = object.getLong("ts");
                        String orderDetailValue = orderDetailState.value();
                        String orderInfoValue = orderInfoState.value();
                        Long orderCount =1L;
                        BigDecimal orderAmount = new BigDecimal(splitFinalAmount);
                        //判断该笔订单详情是否重复
                        if (!id.equals(orderDetailValue)){
                            //该订单详情是第一次来
                            orderDetailState.update(id);
                        }else {
                            //该订单详情不是第一次进来
                            orderAmount = new BigDecimal(0);
                        }

                        //判断order_id是否第一次进来
                        if(!orderId.equals(orderInfoValue)){
                            //order_id第一次进来
                            orderInfoState.update(orderId);
                        }else {
                            orderCount = 0L;
                        }
                        return TradeSourceProvinceUserOrderBean.builder()
                                .stt("")
                                .edt("")
                                .provinceId(provinceId)
                                .sourceId(sourceId)
                                .userId(userId)
                                .orderCount(orderCount)
                                .orderAmount(orderAmount)
                                .ts(ts)
                                .build();
                    }
                });
    }
}
